package com.lihuinch;

import org.fusesource.hawtbuf.Buffer;
import org.fusesource.hawtbuf.UTF8Buffer;
import org.fusesource.mqtt.client.*;

import java.io.IOException;
import java.net.URISyntaxException;
import java.util.Properties;

/**
 * @author lihuinch
 * @date 2019/1/30 16:24
 */
public class MqttTest {

    private final static String topicname = "test";

    private final static String CONNECTION_STRING = "tcp://192.168.13.240:1883";
    private final static boolean CLEAN_START = true;
    private final static String CLIENT_ID = "server";
    private final static short KEEP_ALIVE = 30;// 低耗网络，但是又需要及时获取数据，心跳30s

    public static Topic[] topics = {
            new Topic(topicname, QoS.EXACTLY_ONCE),
            new Topic(topicname, QoS.AT_LEAST_ONCE),
            new Topic(topicname, QoS.AT_MOST_ONCE)};

    public final static long RECONNECTION_ATTEMPT_MAX = 6;
    public final static long RECONNECTION_DELAY = 2000;

    //发送最大缓冲为2M
    public final static int SEND_BUFFER_SIZE = 2 * 1024 * 1024;


    public static void main(String[] args) throws Exception {
       //fun1();
      fun2();


/*
        final CallbackConnection connection = mqtt.callbackConnection();
        connection.listener(new Listener() {

            public void onDisconnected() {
            }

            public void onConnected() {
            }

            public void onPublish(UTF8Buffer topic, Buffer payload, Runnable ack) {
                // You can now process a received message from a topic.
                // Once process execute the ack runnable.
                ack.run();
            }

            public void onFailure(Throwable value) {
                connection.close(null); // a connection failure occured.
            }
        });
        connection.connect(new Callback<Void>() {
            public void onFailure(Throwable value) {

                value.printStackTrace();
                // result.failure(value); // If we could not connect to the server.
            }

            // Once we connect..
            public void onSuccess(Void v) {

                // Subscribe to a topic
                Topic[] topics = {new Topic("foo", QoS.AT_LEAST_ONCE)};
                connection.subscribe(topics, new Callback<byte[]>() {
                    public void onSuccess(byte[] qoses) {
                        // The result of the subcribe request.
                    }

                    public void onFailure(Throwable value) {
                        connection.close(null); // subscribe failed.
                    }
                });

                // Send a message to a topic
                connection.publish("foo", "Hello".getBytes(), QoS.AT_LEAST_ONCE, false, new Callback<Void>() {
                    public void onSuccess(Void v) {
                        // the pubish operation completed successfully.
                    }

                    public void onFailure(Throwable value) {

                        value.printStackTrace();
                        connection.close(null); // publish failed.
                    }
                });

                // To disconnect..
                connection.disconnect(new Callback<Void>() {
                    public void onSuccess(Void v) {
                        // called once the connection is disconnected.
                    }

                    public void onFailure(Throwable value) {
                        // Disconnects never fail.
                    }
                });
            }
        });*/

    }


    public static void fun1() throws Exception {
        Properties prop = new Properties();
        prop.load(MqttTest.class.getResourceAsStream("/apollo.properties"));
        String user = prop.getProperty("apollo.user");
        String password = prop.getProperty("apollo.password");
        String host = prop.getProperty("apollo.host");
        Integer port = Integer.parseInt(prop.getProperty("apollo.port"));

        MQTT mqtt = new MQTT();
        mqtt.setHost(host, port);
        mqtt.setUserName(user);
        mqtt.setPassword(password);


        final FutureConnection connection = mqtt.futureConnection();
        connection.connect();
        int count = 1;
        while (true) {
            count++;
            // 用于发布消息，目前手机段不需要向服务端发送消息
            //主题的内容
            String message = "Hello " + count + " MQTT...";
            String topic = "test";
            connection.publish(topic, message.getBytes(), QoS.AT_LEAST_ONCE,
                    false);
            System.out.println("MQTTFutureServer.publish Message " + "Topic Title :" + topic + " context :" + message);

            Thread.sleep(1000);
        }

    }

    public static void fun2() throws Exception {

        Properties prop = new Properties();
        prop.load(MqttTest.class.getResourceAsStream("/apollo.properties"));
        String user = prop.getProperty("apollo.user");
        String password = prop.getProperty("apollo.password");
        String host = prop.getProperty("apollo.host");
        Integer port = Integer.parseInt(prop.getProperty("apollo.port"));

        MQTT mqtt = new MQTT();

        //==MQTT设置说明
        //设置服务端的ip
        mqtt.setHost(CONNECTION_STRING);
        //连接前清空会话信息 ,若设为false，MQTT服务器将持久化客户端会话的主体订阅和ACK位置，默认为true
        mqtt.setCleanSession(CLEAN_START);
        //设置心跳时间 ,定义客户端传来消息的最大时间间隔秒数，服务器可以据此判断与客户端的连接是否已经断开，从而避免TCP/IP超时的长时间等待
        mqtt.setKeepAlive(KEEP_ALIVE);
        //设置客户端id,用于设置客户端会话的ID。在setCleanSession(false);被调用时，MQTT服务器利用该ID获得相应的会话。
        //此ID应少于23个字符，默认根据本机地址、端口和时间自动生成
        mqtt.setClientId(CLIENT_ID);
        //服务器认证用户名
        //mqtt.setUserName("admin");
        //服务器认证密码
        //mqtt.setPassword("admin");



        /*
             //设置“遗嘱”消息的内容，默认是长度为零的消息
             mqtt.setWillMessage("willMessage");
             //设置“遗嘱”消息的QoS，默认为QoS.ATMOSTONCE
             mqtt.setWillQos(QoS.AT_LEAST_ONCE);
             //若想要在发布“遗嘱”消息时拥有retain选项，则为true
             mqtt.setWillRetain(true);
             //设置“遗嘱”消息的话题，若客户端与服务器之间的连接意外中断，服务器将发布客户端的“遗嘱”消息
             mqtt.setWillTopic("willTopic");
 		*/


        //==失败重连接设置说明
        //设置重新连接的次数 ,客户端已经连接到服务器，但因某种原因连接断开时的最大重试次数，超出该次数客户端将返回错误。-1意为无重试上限，默认为-1
        mqtt.setReconnectAttemptsMax(RECONNECTION_ATTEMPT_MAX);
        //设置重连的间隔时间  ,首次重连接间隔毫秒数，默认为10ms
        mqtt.setReconnectDelay(RECONNECTION_DELAY);
        //客户端首次连接到服务器时，连接的最大重试次数，超出该次数客户端将返回错误。-1意为无重试上限，默认为-1
        //mqtt.setConnectAttemptsMax(10L);
        //重连接间隔毫秒数，默认为30000ms
        //mqtt.setReconnectDelayMax(30000L);
        //设置重连接指数回归。设置为1则停用指数回归，默认为2
        //mqtt.setReconnectBackOffMultiplier(2);

        //== Socket设置说明
        //设置socket接收缓冲区大小，默认为65536（64k）
        //mqtt.setReceiveBufferSize(65536);
        //设置socket发送缓冲区大小，默认为65536（64k）
        mqtt.setSendBufferSize(SEND_BUFFER_SIZE);
        ////设置发送数据包头的流量类型或服务类型字段，默认为8，意为吞吐量最大化传输
        mqtt.setTrafficClass(8);

        //==带宽限制设置说明
        //设置连接的最大接收速率，单位为bytes/s。默认为0，即无限制
        mqtt.setMaxReadRate(0);
        //设置连接的最大发送速率，单位为bytes/s。默认为0，即无限制
        mqtt.setMaxWriteRate(0);


        mqtt.setHost(host, port);
        mqtt.setUserName(user);
        mqtt.setPassword(password);


        //使用回调式API
        final CallbackConnection callbackConnection = mqtt.callbackConnection();
        //连接监听
        callbackConnection.listener(new Listener() {
            //接收订阅话题发布的消息
            @Override
            public void onPublish(UTF8Buffer topic, Buffer body, Runnable ack) {
                System.out.println("=============receive msg================ : " + new String(body.toByteArray()));
                ack.run();
            }

            //连接失败
            @Override
            public void onFailure(Throwable value) {
                System.out.println("===========connect failure===========");
                callbackConnection.disconnect(null);
            }

            //连接断开
            @Override
            public void onDisconnected() {
                System.out.println("====mqtt disconnected=====");
            }

            //连接成功
            @Override
            public void onConnected() {
                System.out.println("====mqtt connected=====");
            }
        });
        //连接
        callbackConnection.connect(new Callback<Void>() {
            //连接失败
            @Override
            public void onFailure(Throwable value) {
                System.out.println("============连接失败：" + value.getLocalizedMessage() + "============");
            }

            // 连接成功
            @Override
            public void onSuccess(Void v) {
                //订阅主题
                Topic[] topics = {new Topic(topicname, QoS.EXACTLY_ONCE)};
                callbackConnection.subscribe(topics, new Callback<byte[]>() {
                    //订阅主题成功
                    @Override
                    public void onSuccess(byte[] qoses) {
                        System.out.println("========订阅成功=======");
                    }

                    //订阅主题失败
                    @Override
                    public void onFailure(Throwable value) {
                        System.out.println("========订阅失败=======");
                        callbackConnection.disconnect(null);
                    }
                });

                //发布消息
                callbackConnection.publish(topicname, ("Hello ").getBytes(), QoS.AT_LEAST_ONCE, false, new Callback<Void>() {
                    @Override
                    public void onSuccess(Void v) {
                        System.out.println("===========消息发布成功============");
                    }

                    @Override
                    public void onFailure(Throwable value) {
                        System.out.println("========消息发布失败=======");
                        callbackConnection.disconnect(null);
                    }
                });

            }
        });

        while (true) {

        }


    }
}
